﻿using Microsoft.Extensions.Logging;
using Orleans;
using System;
using System.Threading.Tasks;
using System.Threading;

namespace IOP.Orleans.MessageHub
{
    /// <summary>
    /// 默认消息监听者
    /// </summary>
    /// <typeparam name="TMessage"></typeparam>
    public class MessageDispatcher<TMessage> : IMessageDispatcher<TMessage>, IMessageObserver<TMessage>
        where TMessage : class
    {
        /// <summary>
        /// 当前客户端
        /// </summary>
        public IGrainFactory Client { get; private set; }
        /// <summary>
        /// 监听者名
        /// </summary>
        public string Name { get; private set; } = nameof(TMessage);

        private bool _IsListened = false;
        /// <summary>
        /// 是否处于监听状态
        /// </summary>
        public bool IsListened { get { return Volatile.Read(ref _IsListened); } }
        /// <summary>
        /// MessageGrain创建的拷贝
        /// </summary>
        public IMessageObserver<TMessage> Observer { get; private set; } = null;
        /// <summary>
        /// 事件，当监听者开始监听时
        /// </summary>
        public event Func<IMessageDispatcher, Task> OnObserverBeginListened = null;
        /// <summary>
        /// 事件，当监听者因为意外停止监听时
        /// </summary>
        public event Func<IMessageDispatcher, Task> OnObserverAbortListened = null;
        /// <summary>
        /// 日志
        /// </summary>
        private readonly ILogger<MessageDispatcher<TMessage>> _Logger;
        /// <summary>
        /// 消息处理函数
        /// </summary>
        private Func<TMessage, Task> _MessageHandle { get; set; }
        /// <summary>
        /// 健康检查
        /// </summary>
        private Func<byte[], Task> _HealthCheckHandle { get; set; }
        /// <summary>
        /// 取消令牌
        /// </summary>
        private CancellationTokenSource _TokenSource = new CancellationTokenSource();
        /// <summary>
        /// 信号控制器
        /// </summary>
        private AutoResetEvent _TaskController = new AutoResetEvent(false);

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="logger"></param>
        public MessageDispatcher(IGrainFactory client, string name, ILogger<MessageDispatcher<TMessage>> logger)
        {
            _Logger = logger;
            Client = client;
            Name = name;
            _HealthCheckHandle = DefaultHealthCheckCallback;
        }
        /// <summary>
        /// 健康检查
        /// </summary>
        /// <returns></returns>
        public virtual void HealthCheck(int period = 30000)
        {
            Task.Factory.StartNew(async () =>
            {
                while (!_TokenSource.IsCancellationRequested)
                {
                    await HealthCheckHandle(new byte[] { 1, 2, 3, 4 });
                    _TaskController.WaitOne(period);
                }
            }, _TokenSource.Token);
        }
        /// <summary>
        /// 停止健康检查
        /// </summary>
        public virtual void StopHealthCheck()
        {
            _TokenSource.Cancel();
            _TaskController.Set();
            _TokenSource = new CancellationTokenSource();
        }
        /// <summary>
        /// 收到消息
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        public virtual void ReceiveMessage(TMessage message)
        {
            _MessageHandle?.Invoke(message);
        }
        /// <summary>
        /// 注册消息处理函数
        /// </summary>
        /// <param name="handle"></param>
        public virtual Task RegistMessageHandle(Func<TMessage, Task> handle)
        {
            _MessageHandle = handle;
            return Task.CompletedTask;
        }
        /// <summary>
        /// 是否存活
        /// </summary>
        public virtual bool ImAlive() => Volatile.Read(ref _IsListened);
        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="message"></param>
        public virtual async Task SendMessage(TMessage message, IMessageObserver<TMessage>[] targetObserver = null)
        {
            try
            {
                var msgGrain = Client.GetGrain<IMessageGrain<TMessage>>(Name);
                await msgGrain.SendMessage(message, targetObserver);
            }
            catch (Exception e)
            {
                _Logger.LogError(e.Message + "\r\n" + e.StackTrace);
                throw e;
            }
        }
        /// <summary>
        /// 发送健康检查报文
        /// </summary>
        /// <param name="data"></param>
        /// <param name="targetObserver"></param>
        /// <returns></returns>
        public async Task SendHealthCheck(byte[] data)
        {
            await HealthCheckHandle(data);
        }
        /// <summary>
        /// 接受健康检查
        /// </summary>
        /// <param name="data"></param>
        public virtual void ReceiveHealthCheck(byte[] data)
        {
            _HealthCheckHandle?.Invoke(data);
        }
        /// <summary>
        /// 注册健康检查函数
        /// </summary>
        /// <param name="handle"></param>
        public virtual void RegistHealthCheckHandle(Func<byte[], Task> handle)
        {
            if (handle == null) throw new ArgumentNullException(nameof(handle));
            _HealthCheckHandle = handle;
        }

        /// <summary>
        /// 默认健康检查回执
        /// </summary>
        /// <param name="data"></param>
        /// <returns></returns>
        private Task DefaultHealthCheckCallback(byte[] data)
        {
            _Logger.LogInformation("Success revice health check message");
            return Task.CompletedTask;
        }
        /// <summary>
        /// 健康检查函数
        /// </summary>
        /// <param name="data"></param>
        private async Task HealthCheckHandle(byte[] data)
        {
            try
            {
                var msgGrain = Client.GetGrain<IMessageGrain<TMessage>>(Name);
                if (Volatile.Read(ref _IsListened))
                {
                    if (!await msgGrain.ImAlive(Observer))
                        await msgGrain.Subscribe(Observer);
                    await msgGrain.SendHealthCheck(new byte[] { 0, 1, 2, 3 }, new IMessageObserver<TMessage>[] { Observer });
                }
                else
                {
                    if (Observer != null)
                    {
                        if (await msgGrain.ImAlive(Observer)) await msgGrain.UnSubscribe(Observer);
                        await Client.DeleteObjectReference<IMessageObserver<TMessage>>(Observer);
                    }
                    Observer = await Client.CreateObjectReference<IMessageObserver<TMessage>>(this);
                    await msgGrain.Subscribe(Observer);
                    Volatile.Write(ref _IsListened, true);
                    await msgGrain.SendHealthCheck(data, new IMessageObserver<TMessage>[] { Observer });
                    OnObserverBeginListened?.Invoke(this);
                }
            }
            catch (Exception e)
            {
                Volatile.Write(ref _IsListened, false);
                _Logger.LogError(e.Message + "\r\n" + e.StackTrace);
                OnObserverAbortListened?.Invoke(this);
            }
        }
    }
}
